/*
 * Copyright (c) Huawei Technologies Co., Ltd. 2022-2022. All rights reserved.
 */

package com.huaweicloud.reqmt;

import com.huaweicloud.reqmt.domain.dws_haydn_self_verification_dura_reqmt_d;
import com.huaweicloud.reqmt.function.DwsMapFunction;
import com.huaweicloud.reqmt.source.DwsSourceDws001;

import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 代码 DWD001
 * 表名 dwd_haydn_case_uuid_vaild_dura_stage_d
 *
 * @since 2022-11-03
 */
public class MainDws001 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        String username = args[0];
        String password = args[1];
        String url = args[2];
        // 添加数据源
        DataStreamSource<dws_haydn_self_verification_dura_reqmt_d> source =
            env.addSource(new DwsSourceDws001(username, password, url));
        // 排除节假日
        SingleOutputStreamOperator<dws_haydn_self_verification_dura_reqmt_d> data = source.map(new DwsMapFunction());
        data.print();
        // addSink 将数据导回DWS
        data.addSink(JdbcSink.sink(
            "insert into dws.dws_haydn_self_verification_dura_reqmt_noh_d VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ? ,?)",
            (ps, table) -> {
                ps.setString(1, table.getDs());
                ps.setString(2, table.getWorkspaceId());
                ps.setString(3, table.getWorkspaceName());
                ps.setString(4, table.getSoluUuid());
                ps.setString(5, table.getSoluName());
                ps.setString(6, table.getReqmtUuid());
                ps.setString(7, table.getReqmtName());
                ps.setString(8, table.getCaseType());
                ps.setString(9, table.getWorkflowStatus());
                ps.setString(10, table.getStartTime());
                ps.setString(11, table.getEndTime());
                ps.setString(12, table.getDuraTrans());
                ps.setString(13, table.getDuraSeconds());
            },
            new JdbcConnectionOptions.JdbcConnectionOptionsBuilder().withUrl(url)
                .withDriverName("org.postgresql.Driver")
                .withUsername(username)
                .withPassword(password)
                .build()));
        // 执行
        env.execute("dws_haydn_self_verification_dura_reqmt_d");
    }
}
